package com.atguigu.bigdata.chapter07.timer;

import com.atguigu.bigdata.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author lzc
 * @Date 2022/9/6 11:23
 */
public class Flink04_Timer_Proctice {
    public static void main(String[] args) throws Exception {
        
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
    
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forMonotonousTimestamps()
                    .withTimestampAssigner((ws, ts) -> ws.getTs())
            )
            .keyBy(WaterSensor::getId)
            .process(new KeyedProcessFunction<String, WaterSensor, String>() {
    
                private long timerTs;
                boolean isFirst = true;
                Integer lastVc = 0;
    
                @Override
                public void onTimer(long timestamp,
                                    OnTimerContext ctx,
                                    Collector<String> out) throws Exception {
                    out.collect(ctx.getCurrentKey() + " 连续5s水位上升, 发出预警....");
                    
                    // 注册新的定时器
                    isFirst = true;  // 吓一跳数据进来, 有会当成第一条数据使用
                }
    
                @Override
                public void processElement(WaterSensor ws,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    // 如何判断是否第一条数据进来: 定义一个变量(boolean值), 默认是true, 第一个来了之后置为false
                    if (isFirst) { // 表示第一条数据进来
                        // 注册定时器
                        timerTs = ws.getTs() + 5000;
                        ctx.timerService().registerEventTimeTimer(timerTs);
                        System.out.println("第一条数据紧进来, 注册一个定时器:" + timerTs);
                        // 把标记置为false
                        isFirst = false;
                    }else{// 不是第一条数据: 要判断水位是否上升
                        // 当前的水位值和上一条水位值做对比
                        if (ws.getVc() < lastVc) {
                            // 水位下降, 取消定时器
                            ctx.timerService().deleteEventTimeTimer(timerTs);
                            System.out.println("水位下降, 取消定时器:" + timerTs);
                            
                            // 注册一个新的定时器
                            timerTs = ws.getTs() + 5000;
                            ctx.timerService().registerEventTimeTimer(timerTs);
                            System.out.println("注册一个新的定时器:" + timerTs);
                        }else{
                            System.out.println("水位上升或不变, 不做任何操作....");
                        }
                    }
    
                    // 把当前的水位存储到变量中, 下一条数据进来的时候判断使用
                    lastVc = ws.getVc();
                }
            })
            .print();
           
        env.execute();
        
    }
}

/*
第一条数据进来的时候, 注册一个5s后触发的定时
如果后面的数据是上升, 则什么都不做

如果出现下降,则删除定时器, 重新注册一个新的5s后触发的定时器
 */